/**
 * 
 */
package com.ws.framework.agent;

import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.shaded.com.google.common.base.Charsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Strings;
import com.google.common.collect.Queues;
import com.ws.framework.util.Constants;
import com.ws.framework.util.PropertiesUtil;
import com.ws.framework.zk.CuratorClient;

/**
 * <p>
 * </p>
 *
 * @author WSH
 */
public class DynamicBeanWithWatchFactory extends AbstractBeanFactory {
	
	final Logger logger = LoggerFactory.getLogger(this.getClass());

	private DynamicBeanWithWatchFactory() {
		Thread back = new Thread(new Runnable() {
			@Override
			public void run() {
				for(;;) {
					try {
						backQueue.take().run();
					} catch (InterruptedException e) {
						// ignore
					}
				}
			}
		});
		back.setDaemon(true);
		back.setName("Watch Source Thread");
		back.start();
	}

	public static final DynamicBeanWithWatchFactory instance = new DynamicBeanWithWatchFactory();

	private CuratorFramework client = new CuratorClient(PropertiesUtil.getValue(Constants.ZOOKEEPER_ADDRESS))
			.getClient();

	private final LinkedBlockingQueue<Runnable> backQueue = Queues.newLinkedBlockingQueue();

	@Override
	protected void addWatch(final String path) {
		try {
			@SuppressWarnings("resource")
			final NodeCache nodeCache = new NodeCache(client, path, false);
			nodeCache.getListenable().addListener(new NodeCacheListener() {
				@Override
				public void nodeChanged() throws Exception {
					if (nodeCache.getCurrentData().getPath().equals(path)) {
						Runnable job = createBackJob(new String(nodeCache.getCurrentData().getData(), Charsets.UTF_8));
						if (!Objects.isNull(job))
							backQueue.offer(job);
					}
				}
			});
			nodeCache.start();
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	private Runnable createBackJob(final String data) {
		return new Runnable() {
			@Override
			public void run() {
				if(Strings.isNullOrEmpty(data))
					return;
				if(data.startsWith("1_")) {
					referesh(data.substring(2));
				} else if (data.startsWith("2_")) {
					remove(data.substring(2));
				} else {
					logger.warn("Incomprehension the zk command: " + data);
				}
			}
		};
	}
}
